package com.atguigu.gmall.realtime.app.dim.function;


import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.constant.GmallConfig;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Map;
import java.util.Set;

public class DimSinkFuntion implements SinkFunction<JSONObject> {
   // {"database":"gmall","data":{"birthday":"1989-05-14","login_name":"ff42vc8bt","create_time":"2020-06-14 11:36:59","name":"孙纨","user_level":"1","id":900},"sink_table":"dim_user_info","type":"bootstrap-insert","table":"user_info","ts":1668650622}

    // 把json对象中的数据提取出来，形成upsert into 语句   ，执行
    @Override
    public void invoke(JSONObject jsonObject, Context context) throws Exception {
         // upsert  into  db.table (字段名1, 字段名2.....)  values(字段值1,字段值2....)
        String sinkTable = jsonObject.getString("sink_table");
        JSONObject dataJsonObj = jsonObject.getJSONObject("data");
        StringBuilder colnamesSB=new StringBuilder();
        StringBuilder colvaluesSB=new StringBuilder();
        Set<Map.Entry<String, Object>> colEntrySet = dataJsonObj.entrySet();
        for (Map.Entry<String, Object> colEntry : colEntrySet) {
            if(colnamesSB.length()>0){
                colnamesSB.append(",");
                colvaluesSB.append(",");
            }
            colnamesSB.append(colEntry.getKey());
            colvaluesSB.append("'"+colEntry.getValue()+"'");
        }

        String upsertSQL="upsert into "+ GmallConfig.DIM_SCHEMA+"."+sinkTable+"("+colnamesSB+")" +
                " values("+colvaluesSB+")";

        //System.out.println(upsertSQL);

        PhoenixUtil.executeSql(upsertSQL);

    }
}
